package cn.ac.iie.ulss.indexer.worker;

import cn.ac.iie.store.hbase.hbaseop.HbaseUtilOp;
import cn.ac.iie.ulss.indexer.metastore.GetMetastoreMessage;
import cn.ac.iie.ulss.indexer.metric.DataIndexerMetricReportor;
import cn.ac.iie.ulss.indexer.monitor.SimpleMonitor;
import cn.ac.iie.ulss.indexer.monitor.StatusUpdater;
import cn.ac.iie.ulss.indexer.mqproducer.RocketMQProducer;
import cn.ac.iie.ulss.indexer.runenvs.GlobalParas;
import cn.ac.iie.ulss.indexer.shuthandler.KillHandler;
import static cn.ac.iie.ulss.indexer.worker.MainIndexer.log;
import iie.metastore.MetaStoreClient;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.logging.Level;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

public class MainIndexer {

    public static Logger log = Logger.getLogger(MainIndexer.class.getName());

    public static void main(String[] args) {
        PropertyConfigurator.configure("log4j.properties");        
        try {
            GlobalParas.docProtocol = Protocol.parse(new File("doc.json"));  
            GlobalParas.docsProtocol = Protocol.parse(new File("docs.json"));
        } catch (IOException ex) {
            log.error("read doc.json failed",ex);
        }
        GlobalParas.utl_docschema = GlobalParas.docProtocol.getType("doc");
        HbaseUtilOp.initHbaseConnetion();
        
        GlobalParas.init();        
        InitEnvCachedpara.initDataType();
        try {
            MetaStoreClient msc = GlobalParas.clientPool.getClient();   //自己的
            List dbs = msc.client.get_all_attributions();
            List<Thread> threads = new ArrayList<Thread>();
            GlobalParas.clientPool.realseOneClient(msc);
            int dbNumPerThread = dbs.size() / GlobalParas.initMetaInfoThreadnum;    //?
            if (dbNumPerThread == 0) {
                dbNumPerThread = 1;
            }
            log.info("the db num is " + dbs.size() + " and db num per thread is " + dbNumPerThread);
            List dbNames = new ArrayList();
            for (int i = 0; i <= dbs.size() - 1; i++) {         //获得table
                dbNames.add(((Database) dbs.get(i)).getName());
                if ((dbNames.size() >= dbNumPerThread) || (i == dbs.size() - 1)) {
                    InitMetastoreInfo initmetastore = new InitMetastoreInfo(dbNames);
                    Thread initmetasinfoThread = new Thread(initmetastore);
                    initmetasinfoThread.start();
                    dbNames = new ArrayList();
                    threads.add(initmetasinfoThread);
                }
            }
            for (Thread t : threads) {
                t.join();
            }
        } catch (Exception ex) {
            log.error(ex, ex);
            return;
        }
        //GlobalParas.docs_protocal = GlobalParas.imd.getDocSchema();
        //InitEnvCachedpara.initSchema();         //从数据库获得schema的名字
        GlobalParas.docsSchema = GlobalParas.docsProtocol.getType("docs");
        GlobalParas.docsReader = new GenericDatumReader<GenericRecord>(GlobalParas.docsSchema);
        GlobalParas.schemaname2Schema.put("docs", GlobalParas.docsSchema);
        log.info("init schema for docs is ok");
        GlobalParas.utl_docschema = GlobalParas.docProtocol.getType("doc");
        //GlobalParas.docsReader = new GenericDatumReader<GenericRecord>(GlobalParas.docsSchema);
        GlobalParas.schemaname2Schema.put("doc", GlobalParas.utl_docschema);
        //InitEnvCachedpara.initMq2Table();       //形成mq2table映射
        InitEnvCachedpara.initIndexinfo();      //初始化lucene的一些参数
        
        GetMetastoreMessage getMetastoreMessage = new GetMetastoreMessage();
        Thread g = new Thread(getMetastoreMessage);
        g.start();     
        WorkerPool.initWorkpool();
        log.info("begin init the statics data emiter ... ");
        //InitEnvCachedpara.initStatVolumeSchema();

        log.info("init the statics data emiter ok");

//        List<String> topics = new ArrayList<String>();
//        String[] oritopics = GlobalParas.rocketConsumTopics.split("[,]");
//        for (String s : oritopics) {
//            topics.add(GlobalParas.hostName + "_" + s);
//        }

        RocketMQProducer.init(GlobalParas.hostName + "_" + GlobalParas.rocketProducerGroupPrefix + "_producergroup", GlobalParas.rocketNameServer);
        try {
            DataIndexerMetricReportor.init();     //producer发送数据
            log.info("the producer for metric is " + DataIndexerMetricReportor.producer);
        } catch (Exception e) {
            log.error(e, e);
        }
        //topic1,topic2 |rocketMinConsumeThread|rocketMaxConsumeThread|clientCallbackExecutorThreads|dispatcherThreadNum
        for (String key : GlobalParas.topicConfig.keySet()) {           
            ArrayList<String> oriTmpTopics = new ArrayList<String>();
            oriTmpTopics.addAll(Arrays.asList(GlobalParas.topicConfig.get(key).split("[|]")[0].split("[,]")));
            String group = GlobalParas.topicConfig.get(key).split("[|]")[1];          
            int rocketMinConsumeThread = Integer.parseInt(GlobalParas.topicConfig.get(key).split("[|]")[2]);
            int rocketMaxConsumeThread = Integer.parseInt(GlobalParas.topicConfig.get(key).split("[|]")[3]);
            int clientCallbackExecutorThreads = Integer.parseInt(GlobalParas.topicConfig.get(key).split("[|]")[4]);
            int dispatcherThreadNum = Integer.parseInt(GlobalParas.topicConfig.get(key).split("[|]")[5]);
            try {
                ArrayBlockingQueue shareBuffer = new ArrayBlockingQueue(GlobalParas.rocketPullDatabufferSize);               
                //consumer拉取数据
                
                GlobalParas.datapuller = new RocketDataPuller(GlobalParas.hostName + "_" + group + "_" + GlobalParas.rocketConsumeGroupPrefix, oriTmpTopics, "*", shareBuffer, rocketMinConsumeThread, rocketMaxConsumeThread, clientCallbackExecutorThreads);
                DataDispatcher dispatcher = new DataDispatcher(shareBuffer);  //共享buffer
                Thread pulldataThread = new Thread(GlobalParas.datapuller);
                pulldataThread.setDaemon(true);
                pulldataThread.start();
                for (int i = 0; i < dispatcherThreadNum; i++) {
                    Thread t = new Thread(dispatcher);
                    t.setName(oriTmpTopics + "_disp_" + i);
                    t.start();
                }
            } catch (Exception e) {
                log.error(e, e);
            }
        }


        log.info("start init the http receive data server ... \n");
        try {
            HttpGetdataServer.init();
            HttpGetdataServer.startup();
        } catch (Exception ex) {
            log.error(ex, ex);
        }

        log.info("init the http server ok,will init the kill handler");
        KillHandler killhandle = new KillHandler();
        killhandle.registerSignal("TERM");
        log.info("init the kill handler done ");

        SimpleMonitor monitor = new SimpleMonitor();
        Thread monitorThread = new Thread(monitor);
        monitorThread.start();

        StatusUpdater statusUpdater = new StatusUpdater();
        Thread statusUpdateThread = new Thread(statusUpdater);
        statusUpdateThread.start();

        boolean isover = false;
        while ((!GlobalParas.isShouldExit.get()) || (!isover)) {
            isover = true;
            try {
                Thread.sleep(20000L);
            } catch (InterruptedException ex) {
            }
            for (IndexControler c : GlobalParas.id2Createindex.values()) {
                if (!c.isAllOver.get()) {
                    isover = false;
                    break;
                }
            }

            if ((GlobalParas.isShouldExit.get()) && (isover)) {
                log.info("the system can exit safely,will destroy the producer and work pool，wait ... ");
                try {
                    WorkerPool.shutWorkpool();
                } catch (Exception e) {
                    log.error(e, e);
                }

                DataIndexerMetricReportor.shutdown();
                GlobalParas.isUpdaterShouldExit.set(true);

                log.info("shut the rocket producer ok ...");
            }
        }
    }
}